package leo.peersim.paxos;

import java.util.Vector;

import leo.peersim.common.LamportEventComparator;
import leo.peersim.common.LogicalClock;

import org.apache.log4j.Logger;

import peersim.core.CommonState;
import peersim.edsim.EDSimulator;

/**
 * Proposer agent
 * 
 * @author leo
 * @version June 20, 2009
 */
public class Proposer extends PaxosAgent {
	
	private Logger logger = Logger.getLogger(this.getClass().getName());
	
	private String name = "Proposer";
	
	private PromisedBallots promises = new PromisedBallots();
	
	private Vector<ClientRequest> pendingRequests;
	private Vector<ClientRequest> processedRequests;
	
	private Vector<IClientRequest> _allRequests;

	private LogicalClock _votazioneInCorso;
	
	private ClientRequest _currentClientRequest;

	@Deprecated
	private long _proposingValue;
	
	/**
	 * Leader agent
	 * 
	 * @param container
	 */
	protected Proposer(Paxos container) {
		super(container);
		this.pendingRequests = new Vector<ClientRequest>();
		this.processedRequests = new Vector<ClientRequest>();
		this._allRequests = new Vector<IClientRequest>();
	}
	
	@Override
	protected String getName() {
		return this.name;
	}
	
	/**
	 * <p>If the Leader is already managing a request, it enqueues ClientRequests
	 * to pendingRequests queue</p>
	 * 
	 * @param req
	 */
	protected void submit(ClientRequest req) {
		this.pendingRequests.add(req);
		
		this._allRequests.add(req);
		
		if(this._currentClientRequest==null) {
			this.startManagingNextRequest();
		} else if(this._currentClientRequest.hasReachedConsensus()) {
			this.startManagingNextRequest();
		} else {
			logger.debug(pp() + "request _" + req.getOpName() + "_ enqueued...");
		}
	}

	/**
	 * 
	 */
	private void startManagingNextRequest() {		
		this._currentClientRequest = this.pendingRequests.remove(0);
		this.processedRequests.add(this._currentClientRequest);
		this.triggerNextRound();
	} 
	
	/**
	 * 
	 * @return
	 */
	//FIXME fast paxos! requires to add a sequence number within Accept message
	//and to modify the Acceptor's and Learner's code
	private boolean doesIBelieveImTheLeader() {
//		if(this._votazioneInCorso==null) {
//			return false;
//		}
		
//		if (container._estimateOfRecentLeader.getID()==CommonState.getNode().getID()) {
//			return true;
//		}
		return false;
	}

	/**
	 * 
	 */
	private void phase1() {
		this._votazioneInCorso = container.getNewBallot();
		EDSimulator.add(container._timeout, new Timeout(this._votazioneInCorso), CommonState.getNode(), CommonState.getPid());
		//logger.info(pp() + "scheduling a timeout with delay " + container._timeout);
		
		this.promises = new PromisedBallots();		
		
		logger.debug(pp() + "ballot" + this._votazioneInCorso + ": trying to become leader");
		
		Prepare prepare = new Prepare(CommonState.getNode(), container.getLogicalClock(), this._votazioneInCorso);
		container.broadcastToGroup(prepare);
	}
	
	/**
	 * 
	 */
	private void phase2() {
		//this._votazioneInCorso = container.getNewBallot();
		
		this._currentClientRequest.newAttempt();
		
		long proposedSequenceNumber = this.getASequenceNumberProposal();
		
		logger.debug(pp() + "ballot" + this._votazioneInCorso + ": proposing _" + this._currentClientRequest.getOpName() + "_ (attempt " + this._currentClientRequest.getAttempts() + ") as op " + proposedSequenceNumber);
		
		Accept accept = new Accept(CommonState.getNode(), container.getLogicalClock(), this._votazioneInCorso, this._currentClientRequest.getOpName(), proposedSequenceNumber);
		container.broadcastToGroup(accept);
	}
	
	private long getASequenceNumberProposal() {
		return this._proposingValue;
	}

	/**
	 * 
	 */
	private void triggerNextRound() {
		
		if (!this._currentClientRequest.hasReachedConsensus()) {
			
			if(!this.doesIBelieveImTheLeader()) {
				this.phase1();
			} else {
				this.phase2();
			}
			
		} else {
			//trigger next client request if exists
			if (!this.pendingRequests.isEmpty()) {
				//logger.debug(pp() + "start new request from my queue...");
				this.startManagingNextRequest();
			} else {
				logger.trace(pp() + "sleeping till next request...");
				// idle
			}
		}
	}
	
	/**
	 * 
	 * @param p
	 */
	protected void manage(Promise p) {
		logger.trace(pp()+"received promise from " + p.getSenderNode().getID() + ", latest value: "+ p.getLatestAcceptedValue());
		
		LamportEventComparator c = new LamportEventComparator();
		if (c.compare(p.getBallotId(), this._votazioneInCorso)!=0) {
			// received a promise from an old ballot, dropping
			//TODO
			return;
		}
		
		if (!this.promises.containsKey(p.getSenderNode())){
			
			if (!promises.hasQuorum()) {
				// collect the new promise and re-check if the quorum is reached
				this.promises.put(p.getSenderNode(), p);
				if (promises.hasQuorum()) {
					
					//FIXME is it correct?
					this._proposingValue = promises.getLargestValue();
					//this._proposingValue = promises.getSmallerValue();
					
					logger.debug(pp() + "I'm the new leader!");
					this.phase2();
				}
			} else {
				// it has already a quorum, dropping message!
				//TODO
			}
			
		} else {
			//there is already a proposal for that node
			//TODO
		}
	}
	
	/**
	 * 
	 * @param r
	 */
	protected void manage(Reject r) {
		/*
		logger.debug(pp() + "received a reject for ballot" + r.getBallotId() + " try again with more than " + r.getLatestAcceptedPhaseABallot());
		this.triggerNextRound();
		*/
	}
	
	/**
	 * 
	 * @param r
	 */
	protected void manage(Refuse r) {
		/*
		logger.debug(pp() + "received a refuse for ballot" + r.getBallotId() + " try again with more than " + r.getLatestAcceptedPhaseABallot());
		this.triggerNextRound();
		*/
	}

	/**
	 * 
	 * @param ballotId
	 * @param r
	 */
	protected void decide(LogicalClock ballotId, ReplicaCommand r) {
		if(ballotId.equals(this._votazioneInCorso)){
			this._currentClientRequest.setReplicaCommand(r);
			this.triggerNextRound();
		} else {
			logger.fatal(pp() + "received a decide for a ballot that was not from me! votazioneInCorso:" + this._votazioneInCorso + ", ricevuto: " + ballotId);
		}
	}

	/**
	 * 
	 * @param t
	 */
	protected void timeout(Timeout t) {
		if (t.getTimedOutBallot().equals(this._votazioneInCorso)) {
			//logger.info(pp() + "ballot:" +this._votazioneInCorso + " time out...");
			this.triggerNextRound();
		}
	}
	
	/**
	 * 
	 * @return
	 */
	protected Vector<IClientRequest> getAllClientRequests() {
		return this._allRequests;
	}
}

class Timeout {

	private LogicalClock _ballot;
	
	public Timeout(LogicalClock ballot) {
		this._ballot = ballot;
	}
	
	public LogicalClock getTimedOutBallot() {
		return this._ballot;
	}

}